package streamWorker

import (
	"fmt"
	"gitee.com/kingzyt/common/log"
	"gitee.com/kingzyt/common/util"
	"io"
	"sync"
	"sync/atomic"
)

type CloseType int32

const (
	Close_type_abort CloseType = iota
	Close_type_finishall
)

const (
	status_closed int32 = iota
	status_running
	status_reader_closing
	status_proc_closing
)

const default_chan_size int = 64
const default_bulk_size int = 128

/*
单个的chan是一个生产者消费者的队列，能达到很好的运转效率，
但是chan中使用的元素没有被循环利用，
因为高流量的请求处理需要在不大量申请和释放内存，造成内存碎片的前提下，
保证运转效率不降低。所以这里使用两个chan，一个是工作chan，一个是回收再利用chan，
reader不断从回收chan取空间填入数据后，加入工作chan；
proc从工作chan取数据，完成操作后返还空间给回收chan
*/
type StreamWorker struct {
	name              string
	workChan          chan []byte
	recoveryChan      chan []byte
	chanSize          int
	bulks             [][]byte
	status            int32
	closeType         CloseType
	closeSign         sync.WaitGroup
	changeStatusMutex sync.Mutex

	onStoppedAfter func()

	// for debug // pos0 int32
	// for debug // pos1 int32
	// for debug // pos2 int32
}

// procFunc中不应该有goroutine，也不应该将获得的data传递到外部使用
func New(name string, chanSize int, bulkSize int, closeType CloseType) *StreamWorker {
	if chanSize < 0 {
		chanSize = default_chan_size
	}
	if bulkSize <= 0 {
		bulkSize = default_bulk_size
	}

	rlt := &StreamWorker{
		name:      name,
		chanSize:  chanSize,
		bulks:     make([][]byte, chanSize),
		closeType: closeType,
	}

	// bulk可被重用
	for i := 0; i < chanSize; i++ {
		rlt.bulks[i] = make([]byte, bulkSize)
	}
	return rlt
}

type StopParams struct {
	OnStoppedAfter func()
}

func (self *StreamWorker) SetStopParams(params StopParams) {
	self.onStoppedAfter = params.OnStoppedAfter
}

type StartParams struct {
	OnStartOK      func() // 保证在开始处理前执行准备操作
	Reader         io.Reader
	DataProc       func(data []byte)
	ReadEOFProc    func()
	ReadErrProc    func(err error)
	ReadBeforeProc func()
	ReadAfterProc  func()
}

/*对于status改变，没有同时写的时候，
Start:			closed->running,
Stop:			running->reader_closing,
ReaderClosed:	reader_closing->proc_closing,
ProcClosed:		proc_closing->closed
而这里的状态流转，每一步对于上一步状态的改变不要求立即反应，只要稍后能知道上一步状态发生了变化即可，
状态变化是一步接一步，没有状态的乱序变化，状态流转是固定的
这里没有任何数据与status关联，除了chan是在goroutine间使用的，但它是goroutine安全的，没有其他数据在goroutine间共享，
bulk因为在chan中，所以也在任意时刻仅属于一个goroutine，不会出现共享使用bulk的现象，
不需要status保护线程安全，任何数据都不需要在状态发生改变的当下做出反应
status只是起到一个提示状态变化的功能，由每一步自行判定并流转状态，
这样就不用因为使用锁来保护大段代码而影响性能，就直接在status上使用atomic保护status本身的原子性即可

read和proc的for不能用锁直接lock，因为是一个无限循环，会导致其它用锁的goroutine被直接阻塞

【这里唯一的约定是，reader必须完全委托这里，不能再在外部其他的goroutine中做Read操作，否则会影响到这里数据读取的完整性】
*/
func (self *StreamWorker) Start(startParams StartParams) bool {
	if !self.changeStatus(status_closed, status_running, func() {
		// work line & read data line both closed, then the end could be arrived
		self.closeSign.Add(2)
		/*重启会新建chan，空间代价较小，因为只有close chan才能正确处理关闭，而且chan没有重启操作，
		但重用bulk，最大限度复用空间*/
		self.workChan = make(chan []byte, self.chanSize)
		self.recoveryChan = make(chan []byte, self.chanSize)

		/*本来reader的速度就是限制chan间流转的关键，有数据进来才能让proc有数据处理，
		reader流转越快，proc处理越快，不会因为recoveryChan满了而阻塞workChan，
		因为这个时候workChan为空，recoveryChan已满，所有数据已经处理完毕，没有数据需要处理*/
		for _, bulk := range self.bulks {
			self.recoveryChan <- bulk
		}
	}, nil) {
		return false
	}

	reader := startParams.Reader
	if reader == nil {
		panic(fmt.Errorf("%s: Reader is nil", self.name))
	}
	proc := startParams.DataProc
	if proc == nil {
		panic(fmt.Errorf("%s: DataProc is nil", self.name))
	}
	readEOFProc := startParams.ReadEOFProc
	readErrProc := startParams.ReadErrProc
	readBeforeProc := startParams.ReadBeforeProc
	readAfterProc := startParams.ReadAfterProc

	onStartOK := startParams.OnStartOK
	if onStartOK != nil {
		onStartOK()
	}

	// reader bulk <- recoveryChan, stream -> data, data -> bulk, bulk -> workChan
	go func() {
		defer func() {
			// for debug // atomic.StoreInt32(&self.pos0, 9)
			close(self.workChan)

			// for debug // atomic.StoreInt32(&self.pos0, 10)
			if !self.changeStatus(status_reader_closing, status_proc_closing, func() {
				// for debug // atomic.StoreInt32(&self.pos0, 11)
				self.closeSign.Done()
				// for debug // atomic.StoreInt32(&self.pos0, 12)
			}, nil) {
				log.Error(self.name, "status err, need status_reader_closing, now:%d", atomic.LoadInt32(&self.status))
				return
			}

			log.Info(self.name, log.I, "read stopped")
		}()

		for bulk := range self.recoveryChan {
			// for debug // atomic.StoreInt32(&self.pos0, 1)
			if atomic.LoadInt32(&self.status) == status_reader_closing {
				// for debug // atomic.StoreInt32(&self.pos0, 2)
				break
			}

			// for debug // atomic.StoreInt32(&self.pos0, 3)
			if readBeforeProc != nil {
				readBeforeProc() // e.g. setdeadline
			}
			//恢复可能的空间限定
			bulk = bulk[:cap(bulk)]
			n, err := reader.Read(bulk)
			if err != nil {
				if err == io.EOF {
					if readEOFProc != nil {
						readEOFProc()
					}
					self.Stop()
					// for debug // atomic.StoreInt32(&self.pos0, 4)
					break
				}
				if readErrProc != nil {
					readErrProc(err)
				}
				self.Stop()
				// for debug // atomic.StoreInt32(&self.pos0, 5)
				break
			}
			// for debug // atomic.StoreInt32(&self.pos0, 6)
			if readAfterProc != nil {
				readAfterProc()
			}
			//虽然可能限定使用空间，但底层空间不会减小
			bulk = bulk[:n]

			// for debug // atomic.StoreInt32(&self.pos0, 7)
			self.workChan <- bulk
			// for debug // atomic.StoreInt32(&self.pos0, 8)
		}
	}()

	// proc bulk <- workChan, bulk -> data, proc, bulk -> recoveryChan
	go func() {
		defer func() {
			/* proc的goroutine不会进行close workChan，只会在status为status_proc_closing时才退出，
			而status_proc_closing状态，是reader所在goroutine退出时触发改变的
			close workChan只会在reader所在goroutine退出时进行，
			所以这个时候，workChan已经关闭了，我们需要清空workChan中可能的残留，
			需要处理的workChan的bulk，都已经在退出前处理了，有残留的情况必定是close_type_abort，
			无论如何，回收所有的工作bulk*/
			//此时不会有新bulk进入workChan或recoveryChan，也没有针对它们的处理，直接倾倒掉
			// for debug // atomic.StoreInt32(&self.pos1, 8)
			for _ = range self.workChan {
			}
			//先关闭再清空，否则会阻塞在清空这里
			// for debug // atomic.StoreInt32(&self.pos1, 9)
			close(self.recoveryChan)
			// for debug // atomic.StoreInt32(&self.pos1, 10)
			for _ = range self.recoveryChan {
			}

			log.Info(self.name, log.I, "proc stopped")

			// for debug // atomic.StoreInt32(&self.pos1, 10)
			self.closeSign.Done()
			// for debug // atomic.StoreInt32(&self.pos1, 12)
		}()

	ProcEnd:
		for bulk := range self.workChan {
			// for debug // atomic.StoreInt32(&self.pos1, 1)
			if atomic.LoadInt32(&self.status) == status_proc_closing {
				// for debug // atomic.StoreInt32(&self.pos1, 2)
				switch self.closeType {
				case Close_type_abort:
					// for debug // atomic.StoreInt32(&self.pos1, 3)
					break ProcEnd
				case Close_type_finishall:
					// for debug // atomic.StoreInt32(&self.pos1, 4)
					break
				}
			}

			// for debug // atomic.StoreInt32(&self.pos1, 5)
			self.safeProc(proc, bulk)

			// for debug // atomic.StoreInt32(&self.pos1, 6)
			self.recoveryChan <- bulk
			// for debug // atomic.StoreInt32(&self.pos1, 7)
		}
	}()

	return true
}

func (self *StreamWorker) safeProc(proc func(data []byte), bulk []byte) {
	defer func() {
		if r := recover(); r != nil {
			log.Error(self.name, "panic happen:%s", r)
			util.PrintPanicTrace()
		}
	}()

	proc(bulk)
}

// Start之后才可以调用；这里保证在status_running的时候，无论调用多少次Stop，只有一次会真正执行
func (self *StreamWorker) Stop() bool {
	// for debug // atomic.StoreInt32(&self.pos2, 1)
	if !self.changeStatus(status_running, status_reader_closing, nil, nil) {
		// for debug // atomic.StoreInt32(&self.pos2, 2)
		return false
	}

	go func() {
		// for debug // atomic.StoreInt32(&self.pos2, 3)
		self.closeSign.Wait()

		// for debug // atomic.StoreInt32(&self.pos2, 5)
		if self.changeStatus(status_proc_closing, status_closed, nil, nil) {
			// for debug // atomic.StoreInt32(&self.pos2, 6)
			if self.onStoppedAfter != nil {
				self.onStoppedAfter()
			}
		} else {
			log.Error(self.name, "status err, need status_proc_closing, now:%d", atomic.LoadInt32(&self.status))
		}
		// for debug // atomic.StoreInt32(&self.pos2, 7)
	}()

	return true
}

func (self *StreamWorker) GetStatus() int32 {
	return atomic.LoadInt32(&self.status)
}

func (self *StreamWorker) GetWaitProcCount() int {
	return len(self.workChan)
}

func (self *StreamWorker) GetPosForDebug() (int32, int32, int32) {
	// for debug // return atomic.LoadInt32(&self.pos0), atomic.LoadInt32(&self.pos1), atomic.LoadInt32(&self.pos2)
	return -1, -1, -1
}

// 该函数内部不能加入繁重操作,不能有锁相关操作,且不能开放包含外部函数,因为外部函数可能调用其它含changeStatus的sw函数,造成死锁
// 这里主要是用于保证changeSign的操作与状态改变一致
func (self *StreamWorker) changeStatus(from int32, to int32, onOk func(), onFail func()) bool {
	self.changeStatusMutex.Lock()
	defer self.changeStatusMutex.Unlock()

	rlt := atomic.CompareAndSwapInt32(&self.status, from, to)
	if rlt {
		if onOk != nil {
			onOk()
		}
	} else {
		if onFail != nil {
			onFail()
		}
	}
	return rlt
}
